package org.example.task;


import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;


@Slf4j
//@Component
public class RocketMQProducer {
    private static final String TOPIC = "cuntian-study-rocketmq";

    private final RocketMQTemplate rocketMQTemplate;

    public RocketMQProducer(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    /**
     * <h2>使用同步的方式发送消息, 不指定 key 和 tag</h2>
     * */
    public void sendMessageWithValue(String value) {

        // 随机选择一个 Topic 的 Message Queue 发送消息
        SendResult sendResult = rocketMQTemplate.syncSend(TOPIC, value);
        log.info("sendMessageWithValue result: [{}]", JSON.toJSONString(sendResult));

        // 当你使用 rocketMQTemplate.syncSendOrderly(...) 方法发送消息时，
        // "Qinyi" 被用作分组键。RocketMQ 会根据这个键将消息发送到同一个队列中

        SendResult sendResultOrderly = rocketMQTemplate.syncSendOrderly(
                TOPIC, value, "cuntian"
        );
        log.info("sendMessageWithValue orderly result: [{}]",
                JSON.toJSONString(sendResultOrderly));
    }

    /**
     * <h2>使用异步的方式发送消息, 指定 key</h2>
     * key 是用来标识消息的，但它并不是全局唯一的。这意味着多个消息可以具有相同的 key
     * 业务场景：多个消息可以共享相同的 key，特别是在与同一业务实体（如用户、订单等）相关的情况下。这有助于对相关消息进行分组和处理。
     * 顺序消费：在有序发送的场景下，使用相同的 key 可以确保这些消息按顺序被消费
     * */
    public void sendMessageWithKey(String key, String value) {

        Message<String> message = MessageBuilder.withPayload(value)
                .setHeader(RocketMQHeaders.KEYS, key).build();

        // 异步发送消息, 并设定回调
        rocketMQTemplate.asyncSend(TOPIC, message, new SendCallback() {

            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("sendMessageWithKey success result: [{}]",
                        JSON.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable e) {
                log.error("sendMessageWithKey failure: [{}]", e.getMessage(), e);
            }
        });
    }

    /**
     * <h2>使用同步的方式发送消息, 带有 tag, 且发送的是 Java Pojo</h2>
     * 在 RocketMQ 中，tag 的作用主要是用于对消息进行分类和过滤
     * */
    public void sendMessageWithTag(String tag, String value) {

        SendResult sendResult = rocketMQTemplate.syncSend(
                String.format("%s:%s", TOPIC, tag),
                value
        );
        log.info("sendMessageWithTag result: [{}]", JSON.toJSONString(sendResult));
    }


    /**
     * <h2>使用同步的方式发送消息, 带有 key 和 tag</h2>
     * */
    public void sendMessageWithAll(String key, String tag, String value) {

        Message<String> message = MessageBuilder.withPayload(value)
                .setHeader(RocketMQHeaders.KEYS, key).build();
        SendResult sendResult = rocketMQTemplate.syncSend(
                String.format("%s:%s", TOPIC, tag),
                message
        );
        log.info("sendMessageWithAll result: [{}]", JSON.toJSONString(sendResult));
    }

}
